package flink.api_study.state;

import flink.api_study.source.Person;
import flink.api_study.source.PersonSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Collections;
import java.util.List;

/**
 * Created by tzs on 2022/8/2.
 */
public class KeyStateTest {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);

        DataStream<Person> stream = env.addSource(new PersonSource());

        stream.keyBy(person -> person.getSex())
                .map(new MyMapFuction1())
                .print();
        stream.print();
        env.execute();
    }

    static class MyMapFuction1 extends RichMapFunction<Person, Integer> {
        private ValueState<Integer> keyCountState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count",Integer.class,0));
        }

        @Override
        public Integer map(Person person) throws Exception {
            Integer count = keyCountState.value();
            count++;
            keyCountState.update(count);
            return count;
        }
    }
}